package com.handy.socket;

import java.util.LinkedList;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.nio.channels.Selector;
import java.nio.channels.SelectionKey;
import java.util.Iterator;
import java.util.Set;

import org.apache.log4j.Logger;


/**
 * 主控服务线程
 * 
 * @author rocken.zeng@gmail.com
 * @version 1.0
 */

public class Server implements Runnable {
	private static LinkedList<SelectionKey> wpool = new LinkedList<SelectionKey>(); // 回应池

	private static Selector selector;

	private static int MAX_THREADS = 4;

	private ServerSocketChannel sschannel;

	private InetSocketAddress address;

	private Notifier notifier;

	private String ip;

	private int port;

	public Server() {
	}

	/**
	 * 创建主控服务线程
	 * 
	 * @param port
	 *            服务端口
	 * @throws java.lang.Exception
	 */
	public void initialize(String ip, int port) {
		this.ip = ip;
		this.port = port;

		// 获取事件触发器
		notifier = Notifier.getNotifier();
		startRWThread();
		try {
			// 创建无阻塞网络套接
			selector = Selector.open();
			sschannel = ServerSocketChannel.open();
			sschannel.configureBlocking(false);
			address = new InetSocketAddress(ip, port);
			ServerSocket ss = sschannel.socket();
			ss.bind(address);
			sschannel.register(selector, SelectionKey.OP_ACCEPT);
		} catch (IOException e) {
			log.error(e.getMessage());
		}
	}

	protected void startRWThread() {
		// 创建读写线程池
		for (int i = 0; i < MAX_THREADS; i++) {
			Thread r = new Reader();
			Thread w = new Writer();
			r.start();
			w.start();
		}
	}

	public void run() {
		log.info("服务器启动...");
		if (log.isDebugEnabled()){
			log.info("监听地址："+ip+":"+port);
		}
		// 监听
		while (true) {
			int num = 0;
			try {
				num = selector.select();
			} catch (Exception csx) {
				if (selector.isOpen())
					continue;
				else {
					log.error(csx.getMessage());
					break;
				}
			}
			if (num > 0) {
				Set<SelectionKey> selectedKeys = selector.selectedKeys();
				Iterator<SelectionKey> it = selectedKeys.iterator();
				while (it.hasNext()) {
					SelectionKey key = it.next();
					it.remove();
					if (!key.isValid())
						continue;
					try {
						// 处理IO事件
						if ((key.readyOps() & SelectionKey.OP_ACCEPT) == SelectionKey.OP_ACCEPT) {
							// Accept the new connection
							ServerSocketChannel ssc = (ServerSocketChannel) key
									.channel();
							notifier.fireOnAccept();

							SocketChannel sc = ssc.accept();
							sc.configureBlocking(false);

							// 触发接受连接事件
							Request request = new Request(sc);
							notifier.fireOnAccepted(request);

							// 注册读操作,以进行下一步的读操作
							sc
									.register(selector, SelectionKey.OP_READ,
											request);
						} else if ((key.readyOps() & SelectionKey.OP_READ) == SelectionKey.OP_READ) {
							Reader.processRequest(key); // 提交读服务线程读取客户端数据
							key.cancel();
						} else if ((key.readyOps() & SelectionKey.OP_WRITE) == SelectionKey.OP_WRITE) {
							Writer.processRequest(key); // 提交写服务线程向客户端发送回应数据
							key.cancel();
						}
					} catch (Exception exp) {
						key.cancel();
						log.error(exp.getMessage());
					}
				}
			} else {
				addRegister(); // 在Selector中注册新的写通道
			}
		}
	}

	/**
	 * 添加新的通道注册
	 */
	protected void addRegister() {
		synchronized (wpool) {
			while (!wpool.isEmpty()) {
				SelectionKey key = (SelectionKey) wpool.remove(0);
				SocketChannel schannel = (SocketChannel) key.channel();
				try {
					schannel.register(selector, SelectionKey.OP_WRITE, key
							.attachment());
				} catch (Exception e) {
					try {
						schannel.finishConnect();
						schannel.close();
						schannel.socket().close();
						notifier.fireOnClosed((Request) key.attachment());
					} catch (Exception e1) {
					}
					log.error("Error occured in addRegister: "
							+ e.getMessage());
				}
			}
		}
	}

	/**
	 * 提交新的客户端写请求于主服务线程的回应池中
	 */
	
	public static void processWriteRequest(SelectionKey key) {
		synchronized (wpool) {
			wpool.add(wpool.size(), key);
			wpool.notifyAll();
		}
		selector.wakeup(); // 解除selector的阻塞状态，以便注册新的通道
	}
	
	private static Logger log = Logger.getLogger(Server.class);
}
